package com.k.activemq.openwire4.converter;

import javax.jms.*;
import java.io.Serializable;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Map;

public class SimpleMessageConverter implements MessageConverter {

    @Override
    public Message toMessage(Object object, Session session) throws JMSException, RuntimeException {
        if (object instanceof Message) {
            return (Message) object;
        } else if (object instanceof String) {
            return createMessageForString((String) object, session);
        } else if (object instanceof byte[]) {
            return createMessageForByteArray((byte[]) object, session);
        } else if (object instanceof Map) {
            return createMessageForMap((Map<?, ?>) object, session);
        } else if (object instanceof Serializable) {
            return createMessageForSerializable(((Serializable) object), session);
        } else {
            throw new RuntimeException("Cannot convert object of type [" + (object == null ? null : object.getClass().getName()) + "]");
        }
    }

    @Override
    public Object fromMessage(Message message) throws JMSException, RuntimeException {
        if (message instanceof TextMessage) {
            return extractStringFromMessage((TextMessage) message);
        } else if (message instanceof BytesMessage) {
            return extractByteArrayFromMessage((BytesMessage) message);
        } else if (message instanceof MapMessage) {
            return extractMapFromMessage((MapMessage) message);
        } else if (message instanceof ObjectMessage) {
            return extractSerializableFromMessage((ObjectMessage) message);
        } else {
            return message;
        }
    }


    protected TextMessage createMessageForString(String text, Session session) throws JMSException {
        return session.createTextMessage(text);
    }


    protected BytesMessage createMessageForByteArray(byte[] bytes, Session session) throws JMSException {
        BytesMessage message = session.createBytesMessage();
        message.writeBytes(bytes);
        return message;
    }


    protected MapMessage createMessageForMap(Map<?, ?> map, Session session) throws JMSException {
        MapMessage message = session.createMapMessage();
        for (Map.Entry<?, ?> entry : map.entrySet()) {
            Object key = entry.getKey();
            if (!(key instanceof String)) {
                throw new RuntimeException("Cannot convert non-String key of type ["+(key == null ? null : key.getClass().getName())+"]");
            }
            message.setObject((String) key, entry.getValue());
        }
        return message;
    }


    protected ObjectMessage createMessageForSerializable(Serializable object, Session session) throws JMSException {
        return session.createObjectMessage(object);
    }


    protected String extractStringFromMessage(TextMessage message) throws JMSException {
        return message.getText();
    }

    protected byte[] extractByteArrayFromMessage(BytesMessage message) throws JMSException {
        byte[] bytes = new byte[(int) message.getBodyLength()];
        message.readBytes(bytes);
        return bytes;
    }


    @SuppressWarnings("unchecked")
    protected Map<String, Object> extractMapFromMessage(MapMessage message) throws JMSException {
        Map<String, Object> map = new HashMap<>();
        Enumeration<String> en = message.getMapNames();
        while (en.hasMoreElements()) {
            String key = en.nextElement();
            map.put(key, message.getObject(key));
        }
        return map;
    }

    protected Serializable extractSerializableFromMessage(ObjectMessage message) throws JMSException {
        return message.getObject();
    }

}
